package com.kohburg.baseplate.xdelay.factory;

import com.kohburg.baseplate.xdelay.constant.XdelayConstant;
import com.kohburg.baseplate.xdelay.entity.BaseMessage;
import com.kohburg.baseplate.xdelay.service.XdelayMessageService;
import com.kohburg.baseplate.xdelay.store.FlashStore;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;

/**
 * rabbit template工厂
 *
 * @author : zenos hua
 * @date : 2020/6/17 17:51
 **/
@Component
public class RabbitTemplateFactory {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private XdelayMessageService xdelayMessageService;

    private RabbitTemplate instace;

    /**
     * 创建对象
     *
     * @return RabbitTemplate
     */
    public <E> E getInstace(){
        if(instace == null){
            synchronized (RabbitTemplateFactory.class) {
                if(instace == null){
                    instace = rabbitTemplate;
                    instace.setMandatory(true);
                    instace.setConfirmCallback(confirmCallback);
                    instace.setReturnCallback(returnCallback);
                }
            }
        }
        return (E)instace;
    }

    /**
     * 投递交换机失败回调
     */
    private final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            String messageId =  correlationData.getId();;
            try {
                // 消息id
                if(!ack){
                    // 取出消息体
                    Object message = FlashStore.message.get(messageId);

                    // 发送错误信号
                    xdelayMessageService.failedSignal(
                            XdelayConstant.ERROR_TYPE_EXCHANGE,
                            cause,
                            message
                    );
                }
            } finally {
                // 移除内存
                FlashStore.message.remove(messageId);
            }
        }

    };

    /**
     * 投递队列失败回调
     */
    private final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

            BaseMessage baseMessage = (BaseMessage)toObject(message.getBody());
            // 发送错误信号
            xdelayMessageService.failedSignal(
                    XdelayConstant.ERROR_TYPE_QUEUE,
                    replyText,
                    exchange,
                    routingKey,
                    baseMessage.toString()
            );
        }
        private Object toObject(byte[] bytes){
            Object object = null;
            try {
                ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);// 创建ByteArrayInputStream对象
                ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);// 创建ObjectInputStream对象
                object = objectInputStream.readObject();// 从objectInputStream流中读取一个对象
                byteArrayInputStream.close();// 关闭输入流
                objectInputStream.close();// 关闭输入流
            } catch (IOException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
            return object;// 返回对象
        }
    };
}
